<?php

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class Mqtt
{
    public $connection;
    public $channel;
    public $vhostName;
    public $exchangeName;
    public $queueName;
    public $routerKey;

    public function __construct()
    {
        $config = require(__DIR__ . '/config.php');
        $this->connection = new AMQPStreamConnection(
            $config['host'],
            $config['port'],
            $config['user'],
            $config['password'],
            $config['vhost']
        );
    }

    // 创建一个通道：channelId-可选
    public function createChannel()
    {
        $this->channel = $this->connection->channel();
    }

    // 创建一个队列
    public function createQueue($queueName, $passive=false, $durable=false, $exclusive=false, $autoDelete=true, $noWait=false)
    {
        /*
        创建一个队列：
            @param queueName 队列名称，
            @param passive 设置为true表示：如果存在这个队列则返回队列的信息 ，如果不存在这个队列则抛异常
            @param durable 是否持久化（重启后队列还存在） 
            @param exclusive 是否独占队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除
            @param auto_delete 所有消费断开连接后是否自动删除队列
            @param nowait = false 不等待处理结果
        */
        $this->queueName = $queueName;
        $this->channel->queue_declare($this->queueName, $passive, $durable, $exclusive, $autoDelete, $noWait);
    }

    // 创建交换机
    public function createExchange($exchangeName, $type, $passive=false, $durable=false, $autoDelete=false, $internal=false, $nowait=false)
    {
        /**
         * @param $exchange, 交换机名称
         * @param $type, 交换机类型：fanout, direct, topic
         * @param $passive = false, 设置为true表示：如果存在这个交换机则返回交换机的信息 ，如果不存在则抛异常
         * @param $durable = false, 设置true表示将将换机存盘，在服务器重启时不会丢失相关信息
         * @param $auto_delete = true, 设置true表示 自动删除交换机，至少1个队列或交换机与这个交换机绑定对队列解绑
         * @param $internal = false, 设置true表示是内部交换机，客户端不能直接发消息到这个交换机中，只能通过路由
         * @param $nowait = false, 设置true表示不等待处理结果
         */
        $this->exchangeName = $exchangeName;
        $this->channel->exchange_declare(
            $exchangeName,
            $type,
            $passive,
            $durable,
            $autoDelete,
            $internal,
            $nowait
        );
    }

    // 绑定队列和交换机
    public function bindQueueAndExchange($queueName, $exchangeName, $routerKey='', $nowait=false)
    {
        $this->channel->queue_bind($queueName, $exchangeName, $routerKey, $nowait);
    }

    public function bindQE()
    {
        $this->channel->queue_bind($this->queueName, $this->exchangeName, $this->routerKey);
    }
    
    public function setQos($prefetch=1)
    {
        /**
         * 只有consumer已经处理并确认了上一条message时queue才分派新的message给它，
         * RabbitMQ就会把消息分发给下一个空闲的工作者（worker）
         * @param int  $prefetch_size //未实现
         * @param int  $prefetch_count=N //告诉RabbitMQ不要同时给一个消费者推送多于N个消息,一旦有N个消息还没ack,则不在分派新消息给它
         * @param bool $a_global  //未实现
         */
        $this->channel->basic_qos(null, $prefetch, null);
    }

    // 通过通道 向 队列 推送消息
    public function makeMessage($content, $options=[])
    {
        $properties = array_merge([], $options);
        return new AMQPMessage($content, $properties);

        $msg2 = new AMQPMessage("hello", [
            'content_type' => 'text/plain, application/json, text/xml',
            'content_encoding' => '',
            'application_headers' => [],
            'delivery_mode' => 1, //1-不持久化，2-持久化
            'priority' => 99, //优先级别
            'timestamp' => '', //时间错
            'reply_to' => '',
            'expiration' => ''
        ]);
    }

    // 发布消息
    public function publishMsg($content, $option=[], $routerKey='')
    {
        $this->routerKey = $routerKey;
        $msg = $this->makeMessage($content, $option);

        /**
         * @param AMQPMessage $msg //消息内容
         * @param $exchange = '' //交换机名称,
         * @param $routing_key = '' //路由key
         * @param $mandatory = false, // true表示：如果交换机无法把消息发送到具体的队列中，就把消息发送到失败投递记录中，false表示丢弃
         * @param $immediate = false, //已经废弃
         * @param $ticket = null
         */
        $this->channel->basic_publish($msg, $this->exchangeName, $this->routerKey);
    }

    /**
     * @param $queue = '', //队列名称
     * @param $consumer_tag = '', //消费者标签
     * @param $no_local = false, //属于AMQP的标准,但是rabbitMQ并没有做实现.
     * @param $no_ack = false, //false表示收到消息后需要回复确认才被认为被消费
     * @param $exclusive = false, //true表示：这个队列只能由一个消费者消费
     * @param $nowait = false, //true表示不需等待执行结果
     * @param $callback = null, //回调
     */
    public function consumeMsg($callback, $consumerTag='', $noLocal=false, $noAck=true, $exlusive=false, $noWait=false)
    {
        $this->channel->basic_consume($this->queueName, $consumerTag, $noLocal, $noAck, $exlusive, $noWait, 
            function(AMQPMessage $msg) use($callback, $noAck){
                $callback($msg->body);

                if(!$noAck) {
                    // 消息处理完：可以丢弃
                    // false表示只应答tag=8的消息
                    // true表示应答tag=5,6,7,8的消息
                    $msg->ack($multiple=false);
                    // 消息处理失败：不能丢弃
                    $msg->nack();
                    // 消息拒绝：可以丢弃了
                    $msg->reject();
                }
            }
        );

        while ($this->channel->is_open()) {
            $this->channel->wait();
        }
    }

    // 关闭连接
    public function close()
    {
        $this->channel->close();
        $this->connection->close();
    }
}